package com.ylkj.bus.container.report;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson2.JSONObject;
import com.ylkj.bus.MessageContent;
import com.ylkj.bus.app.report.DeviceContainerDO;
import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RdsUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.MessageFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @Author: wengzx
 * @Date: 2025/3/28
 * @Desc: 容器状态上报处理类
 */
@RequiredArgsConstructor
@Slf4j
public class ReportContainerProcess extends ProcessFunction<MessageContent, MessageContent> {


    private final OutputTag<List<DeviceContainerDO>> outTagRds1;


    public static final String TYPE = "type";

    /**
     * 容器状态上报
     */
    public static final String STATUS = "REP_CON_STATUS";

    /**
     * 日期格式，年月日时分秒，年月日用横杠分开，时分秒用冒号分开
     * 例如：2005-05-10 23：20：00，2008-08-08 20:08:08
     */
    public static final String DATE_TIME_FORMAT_YYYY_MM_DD_HH_MI_SS = "yyyy-MM-dd HH:mm:ss";

    @Override
    public void open(Configuration parameters) throws Exception {

    }

    @Override
    public void processElement(MessageContent mqttMsg, Context ctx, Collector<MessageContent> out) throws Exception {
        try {
            JSONObject body = JSONObject.parseObject(mqttMsg.getBody());
            if (!STATUS.equals(body.getString(TYPE))) {
                if (log.isDebugEnabled()) {
                    log.debug("消息类型不匹配: {}", body);
                }
                return;
            }
            String[] topicParts = mqttMsg.getTopic().split("/"); //主题格式:/v1/{}/app/data
            String deviceCode = topicParts.length > 2 ? topicParts[2] : StringUtils.EMPTY;
            if (StringUtils.isBlank(deviceCode)) {
                log.warn("无效的设备编码: {}", mqttMsg.getTopic());
                return;
            }
            processContainerReport(body, deviceCode, ctx);
        }catch (Exception e) {
            log.error("消息处理失败: {}", mqttMsg, e);
        }
    }


    /**
     * 处理容器状态上报消息
     * @param body JSON格式的消息体
     * @param deviceCode 从消息主题解析的设备编码
     * @param ctx Flink处理上下文
     */

    private void processContainerReport( JSONObject body, String deviceCode, Context ctx) {

        ContainerStatusResp resp = body.to(ContainerStatusResp.class);
        if (resp == null || resp.getParam() == null || CollectionUtils.isEmpty(resp.getParam().getContStatus())) {
            log.warn("无效的消息体: {}", body);
            return;
        }
        List<ContainerStatusResp.ContainerStatus> dataParamResultList = resp.getParam().getContStatus();
        List<DeviceContainerDO> deviceContainerDOList = new ArrayList<>(dataParamResultList.size());
        for (ContainerStatusResp.ContainerStatus containerStatus : dataParamResultList) {
            DeviceContainerDO deviceContainerDO = new DeviceContainerDO();
            deviceContainerDO.setId(getCacheKey(deviceCode, containerStatus.getContainer()));
            deviceContainerDO.setContainerName(containerStatus.getContainer());
            deviceContainerDO.setDeviceCode(deviceCode);
            deviceContainerDO.setCreateStatus("1");//设备容器创建状态 (CREATE_STATUS 0：创建中1：成功2：失败)
            deviceContainerDO.setSource(2); //source 1:平台新增 2：终端预置
            deviceContainerDO.setIsRemove("1"); //isRemove 是否删除(0：删除，1：存在) 
            this.convertDeviceContainerData(containerStatus, deviceContainerDO);
            deviceContainerDOList.add(deviceContainerDO);
        }
        ctx.output(outTagRds1,deviceContainerDOList);
    }
    private void convertDeviceContainerData(ContainerStatusResp.ContainerStatus dataParamResult, DeviceContainerDO deviceContainerDO) {
        String state = dataParamResult.getState();
        deviceContainerDO.setImageVersion(dataParamResult.getVersion());
        deviceContainerDO.setContainerStatus(ContainerStatusResp.CONTAINER_STATUS_RESP_STATE_RUNNING.equals(state) ? "1" : "0"); //运行状态0:停止1:启动2:部分启动
        deviceContainerDO.setCpuRate(String.valueOf(dataParamResult.getCpuRate()));
        deviceContainerDO.setMemoryRate(String.valueOf(dataParamResult.getMemUsed()));
        deviceContainerDO.setStorageRate(String.valueOf(dataParamResult.getDiskUsed()));
        deviceContainerDO.setIp(dataParamResult.getIp());
        DateTime parse=null;
        try{
            parse = DateUtil.parse(dataParamResult.getStarted(), DATE_TIME_FORMAT_YYYY_MM_DD_HH_MI_SS);
        }catch (Exception e){
            log.error(e.getMessage(),e);
            parse=new DateTime();
        }
        deviceContainerDO.setStarted(parse);
        deviceContainerDO.setLifeTime(dataParamResult.getLifeTime());
        deviceContainerDO.setImage(dataParamResult.getImage());
    }
    private static String getCacheKey(String deviceCode, String containerName) {
        return String.format("%s_%s", deviceCode, containerName);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
